package com.atguigu.flink.datastramapi.sink;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * Created by Smexy on 2023/2/25
 *
 *      flink没有内置向redis写出的方法。需要额外引入Connector.
 *
 *      Connector: 用于连接flink和外部系统。
 *                      读
 *                      写
 *      flink官方也没有提供Redis连接器，社区组织提供了。
 *
 */
public class Demo1_RedisSink
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction());


        /*
            RedisSink(
                        FlinkJedisConfigBase flinkJedisConfigBase： 封装一个连接池，连接redis
                      RedisMapper<IN> redisSinkMapper:  封装写操作命令
                                                set k1 v1
                                                sadd k1 a ,b,c
                      )
         */
        FlinkJedisPoolConfig flinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder()
            .setDatabase(0)
            .setHost("hadoop104")
            .setPort(6379)
            .setMinIdle(5)
            .setMaxIdle(15)
            .setMaxTotal(20)
            .setTimeout(60000)
            .build();


        ds.addSink(new RedisSink<>(flinkJedisPoolConfig,new MyZsetMapper()));


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }

    /*

        RedisCommandDescription(RedisCommand redisCommand, String additionalKey)
                zset:      zadd key score value
                              additionalKey: key
                hash:
                            hset key , { field:value, field1:value1  }
                            additionalKey: key
        RedisCommandDescription(RedisCommand redisCommand)
                string,list,set


     */
    private static class MyZsetMapper implements RedisMapper<WaterSensor>{

        //描述写操作命令
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.ZADD,"wszset");
        }

        //从数据中抽取value
        @Override
        public String getKeyFromData(WaterSensor data) {
            return data.toString();
        }

        //从数据中抽取score
        @Override
        public String getValueFromData(WaterSensor data) {
            return data.getVc().toString();
        }
    }

    private static class MyHashMapper implements RedisMapper<WaterSensor>{

        //描述写操作命令
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET,"wshash");
        }

        //从数据中抽取field
        @Override
        public String getKeyFromData(WaterSensor data) {
            return data.getId();
        }

        //从数据中抽取value
        @Override
        public String getValueFromData(WaterSensor data) {
            return data.toString();
        }
    }

    private static class MySetMapper implements RedisMapper<WaterSensor>{

        //描述写操作命令
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.SADD);
        }

        //从数据中抽取key
        @Override
        public String getKeyFromData(WaterSensor data) {
            return "wsset";
        }

        //从数据中抽取value
        @Override
        public String getValueFromData(WaterSensor data) {
            return data.toString();
        }
    }

    private static class MyListMapper implements RedisMapper<WaterSensor>{

        //描述写操作命令
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.LPUSH);
        }

        //从数据中抽取key
        @Override
        public String getKeyFromData(WaterSensor data) {
            return "wslist";
        }

        //从数据中抽取value
        @Override
        public String getValueFromData(WaterSensor data) {
            return data.toString();
        }
    }

    private static class MyStringMapper implements RedisMapper<WaterSensor>{

        //描述写操作命令
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.SET);
        }

        //从数据中抽取key
        @Override
        public String getKeyFromData(WaterSensor data) {
            return data.getId();
        }

        //从数据中抽取value
        @Override
        public String getValueFromData(WaterSensor data) {
            return data.toString();
        }
    }
}
